package com.zwh.se.concurrency.completableFuture;

import org.junit.Test;

import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.*;

/**
 * CompletableFuture 实现了 Future 接口和 CompletionStage接口。CompletableFuture相当于一个Task编排工具。
 * 默认情况下CompletableFuture会使用公共的ForkJoinPool线程池.
 * CompletableFuture提供了四个静态方法来创建一个异步操作：
 * runAsync() 以Runnable函数式接口类型为参数，没有返回结果，
 * supplyAsync() 以Supplier函数式接口类型为参数，返回结果类型为U；
 * Supplier接口的 get()是有返回值的(会阻塞)
 */
public class CompletableFutureDemo {
    public static void main(String[] args) throws Exception {
        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(new Runnable() {
            @Override
            public void run() {
                System.out.println("没有返回结果");
            }
        });

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("有返回结果");
            if (new Random().nextInt(10) % 2 == 0) {
                int i = 12 / 0;
            }
            System.out.println("执行结束！");
            return "future返回结果";
        });
        // 任务完成或异常方法完成时执行该方法
        // 如果出现了异常,任务结果为null
        future.whenComplete(new BiConsumer<String, Throwable>() {
            @Override
            public void accept(String t, Throwable action) {
                System.out.println(t+" 执行完成！");
            }
        });
        // 出现异常时先执行该方法
        future.exceptionally(new Function<Throwable, String>() {
            @Override
            public String apply(Throwable t) {
                System.out.println("执行失败：" + t.getMessage());
                return "异常xxxx";
            }
        });
        String s = future.get();
        System.out.println(s);
    }


    /**
     *常用方法
     * 依赖关系：
     * thenApply()：把前面任务的执行结果，交给后面的Function
     * thenCompose()：用来连接两个有依赖关系的任务，结果由第二个任务返回
     * and集合关系：
     * thenCombine()：合并任务，有返回值
     * thenAccepetBoth()：两个任务执行完成后，将结果交给thenAccepetBoth处理，无返回值
     * runAfterBoth()：两个任务都执行完成后，执行下一步操作(Runnable类型任务)
     * or聚合关系：
     * applyToEither()：两个任务哪个执行的快，就使用哪一个结果，有返回值
     * acceptEither()：两个任务哪个执行的快，就消费哪一个结果，无返回值
     * runAfterEither()：任意一个任务执行完成，进行下一步操作(Runnable类型任务)
     * 并行执行：
     * allOf()：当所有给定的 CompletableFuture 完成时，返回一个新的 CompletableFuture
     * anyOf()：当任何一个给定的CompletablFuture完成时，返回一个新的CompletableFuture
     * 结果处理：
     * whenComplete：当任务完成时，将使用结果(或 null)和此阶段的异常(或 null如果没有)执行给定操作
     * exceptionally：返回一个新的CompletableFuture，当前面的CompletableFuture完成时，它也完成，当它异常完成时，给定函数的异常触发这个CompletableFuture的完成
     * 结果消费:
     * thenAccept()：对单个结果进行消费
     * thenAcceptBoth()：对两个结果进行消费
     * thenRun()：不关心结果，只对结果执行Action
     */

    //thenApply()：把前面任务的执行结果，交给后面的Function
    @Test
    public void thenApply(){
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            int result = 100;
            System.out.println("第一次运算：" + result);
            return result;
        }).thenApply(number -> {
            int result = number * 3;
            System.out.println("第二次运算：" + result);
            return result;
        });
    }
    //thenCompose()：用来连接两个有依赖关系的任务，结果由第二个任务返回
    @Test
    public void thenCompose(){
        CompletableFuture<Integer> future = CompletableFuture
                .supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        int number = new Random().nextInt(30);
                        System.out.println("第一次运算：" + number);
                        return number;
                    }
                })
                .thenCompose(new Function<Integer, CompletionStage<Integer>>() {
                    @Override
                    public CompletionStage<Integer> apply(Integer param) {
                        return CompletableFuture.supplyAsync(new Supplier<Integer>() {
                            @Override
                            public Integer get() {
                                int number = param * 2;
                                System.out.println("第二次运算：" + number);
                                return number;
                            }
                        });
                    }
                });
    }
    /**
     * thenApply 和 thenCompose的区别：
     * thenApply转换的是泛型中的类型，返回的是同一个CompletableFuture；
     * thenCompose将内部的CompletableFuture调用展开来并使用上一个CompletableFutre调用的结果在下一步的CompletableFuture调用中进行运算，是生成一个新的CompletableFuture。
     */

    //thenAccept()：对单个结果进行消费,这个接口只有输入，没有返回值。
    @Test
    public void thenAccept(){
        CompletableFuture<Void> future = CompletableFuture
                .supplyAsync(() -> {
                    int number = new Random().nextInt(10);
                    System.out.println("第一次运算：" + number);
                    return number;
                }).thenAccept(number ->
                        System.out.println("第二次运算：" + number * 5));
    }
    //thenAcceptBoth()：对两个结果进行消费,当两个CompletionStage都正常完成计算的时候，就会执行提供的action消费两个异步的结果。无返回值
    @Test
    public void thenAcceptBoth() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> futrue1 = CompletableFuture.supplyAsync(()-> {
            int number = new Random().nextInt(3) + 1;
            try {
                TimeUnit.SECONDS.sleep(number);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("任务1结果：" + number);
            return number;
        });
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(()-> {
            int number = new Random().nextInt(3) + 1;
            try {
                TimeUnit.SECONDS.sleep(number);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("任务2结果：" + number);
            return number;
        });
        futrue1.thenAcceptBoth(future2, new BiConsumer<Integer, Integer>() {
            @Override
            public void accept(Integer x, Integer y) {
                System.out.println("最终结果：" + (x + y));
            }
        });
        TimeUnit.SECONDS.sleep(5);
    }
    //thenRun也是对线程任务结果的一种消费函数，与thenAccept不同的是，thenRun会在上一阶段 CompletableFuture计算完成的时候执行一个Runnable，
    // 而Runnable并不使用该CompletableFuture计算的结果。
    @Test
    public void thenRun(){
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            int number = new Random().nextInt(10);
            System.out.println("第一阶段：" + number);
            return number;
        }).thenRun(() ->
                System.out.println("thenRun 执行"));

    }

    //thenCombine:合并两个线程任务的结果，并进一步处理。有返回值
    @Test
    public void thenCombine() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> future1 = CompletableFuture
                .supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        int number = new Random().nextInt(10);
                        System.out.println("任务1结果：" + number);
                        return number;
                    }
                });
        CompletableFuture<Integer> future2 = CompletableFuture
                .supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        int number = new Random().nextInt(10);
                        System.out.println("任务2结果：" + number);
                        return number;
                    }
                });
        CompletableFuture<Integer> result = future1
                .thenCombine(future2, new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer x, Integer y) {
                        return x + y;
                    }
                });
        System.out.println("组合后结果：" + result.get());
    }
    //applyToEither:两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的转化操作。有返回值
    @Test
    public void applyToEither() throws InterruptedException {
        CompletableFuture<Integer> future1 = CompletableFuture
                .supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        int number = new Random().nextInt(10);
                        try {
                            TimeUnit.SECONDS.sleep(number);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("任务1结果:" + number);
                        return number;
                    }
                });
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = new Random().nextInt(10);
                try {
                    TimeUnit.SECONDS.sleep(number);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务2结果:" + number);
                return number;
            }
        });
        future1.applyToEither(future2, new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer number) {
                System.out.println("最快结果：" + number);
                return number * 2;
            }
        });
        TimeUnit.SECONDS.sleep(11);
    }
    //acceptEither:两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的消费操作。无返回值
    @Test
    public void acceptEither() throws InterruptedException {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = new Random().nextInt(10) + 1;
                try {
                    TimeUnit.SECONDS.sleep(number);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("第一阶段：" + number);
                return number;
            }
        });

        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = new Random().nextInt(10) + 1;
                try {
                    TimeUnit.SECONDS.sleep(number);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("第二阶段：" + number);
                return number;
            }
        });

        future1.acceptEither(future2, new Consumer<Integer>() {
            @Override
            public void accept(Integer number) {
                System.out.println("最快结果：" + number);
            }
        });
        TimeUnit.SECONDS.sleep(10);
    }
    //runAfterEither：两个线程任务相比较，有任何一个执行完成，就进行下一步操作，不关心运行结果。
    @Test
    public void runAfterEither() throws InterruptedException {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = new Random().nextInt(5);
                try {
                    TimeUnit.SECONDS.sleep(number);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务1结果：" + number);
                return number;
            }
        });
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = new Random().nextInt(5);
                try {
                    TimeUnit.SECONDS.sleep(number);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务2结果:" + number);
                return number;
            }
        });
        future1.runAfterEither(future2, new Runnable() {
            @Override
            public void run() {
                System.out.println("已经有一个任务完成了");
            }
        }).join();
        TimeUnit.SECONDS.sleep(10);
    }
    //anyOf:anyOf() 的参数是多个给定的 CompletableFuture，当其中的任何一个完成时，方法返回这个 CompletableFuture。
    @Test
    public void anyOf() throws ExecutionException, InterruptedException {
        Random random = new Random();
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(random.nextInt(5));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        });
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(random.nextInt(1));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "world";
        });
        CompletableFuture<Object> result = CompletableFuture.anyOf(future1, future2);
        Object o = result.get();
        System.out.println(o);
    }
    //allOf:allOf方法用来实现多 CompletableFuture 的同时返回。
    @Test
    public void allOf(){
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("future1完成！");
            return "future1完成！";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("future2完成！");
            return "future2完成！";
        });
        CompletableFuture<Void> combindFuture = CompletableFuture.allOf(future1, future2);
        try {
            //两个方法都完成了才会停止
            combindFuture.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}
